import com.google.gson.JsonObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.SaslConfigs;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.util.Date;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.zip.ZipFile;

/**
 * OBU数据处理，使用H2嵌入式数据库替代MySQL
 */
public class ObuProcessor {

    private static final int[] TEA_KEY = new int[]{-327802692, -1412365481, 596647509, -1729066424};

    // 配置信息
    private static String wifiFilePath;
    private static String workFilePath;
    private static int isSkipHistory;
    private static int isBakFile;
    private static String bakFilePath;
    private static String kafkaServer;
    private static int isLogPushData;
    private static String logPushDataPath;

    // 系统常量
    private static final String CONFIG_FILE = "./obu.cfg";

    // 地理坐标转换常量
    private static final double X_PI = 3.141592653589793 * 3000.0 / 180.0;
    private static final double PI = 3.141592653589793;
    private static final double A = 6378245.0;
    private static final double EE = 0.006693421622965943;

    // 系统状态
    private static volatile int threadExit = 0;
    private static final AtomicInteger errorNo = new AtomicInteger(0);

    // 设备字典
    private static final Map<String, DeviceInfo> devDict = new ConcurrentHashMap<>();

    // 数据库连接
    private static Connection h2Connection;

    // 日志记录器
    private static final PrintWriter logger = createLogger();

    /**
     * 设备信息类
     */
    static class DeviceInfo {
        int id;

        String stationNo;

        String devNo;

        public String longitude;

        public String latitude;

        public DeviceInfo(int id, String stationNo, String devNo, String longitude, String latitude) {
            this.id = id;
            this.stationNo = stationNo;
            this.devNo = devNo;
            this.longitude = longitude;
            this.latitude = latitude;
        }
    }

    /**
     * 创建日志记录器
     */
    private static PrintWriter createLogger() {
        try {
            return new PrintWriter(new FileWriter("error.txt", true));
        } catch (IOException e) {
            System.err.println("Failed to create logger: " + e.getMessage());
            return new PrintWriter(System.err);
        }
    }

    /**
     * 记录错误信息
     */
    private static void logError(String message) {
        String timestamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        String logMessage = String.format("%s - %s - %d - ERROR - %s",
                timestamp, Thread.currentThread().getStackTrace()[2].getMethodName(),
                Thread.currentThread().getStackTrace()[2].getLineNumber(), message);
        logger.println(logMessage);
        logger.flush();
    }

    /**
     * 记录信息
     */
    private static void logInfo(String message) {
        String timestamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        String logMessage = String.format("%s - %s - %d - INFO - %s",
                timestamp, Thread.currentThread().getStackTrace()[2].getMethodName(),
                Thread.currentThread().getStackTrace()[2].getLineNumber(), message);
        logger.println(logMessage);
        logger.flush();
    }

    /**
     * 读取配置文件
     */
    private static void loadConfig() {
        Properties props = new Properties();
        try (FileInputStream fis = new FileInputStream(CONFIG_FILE)) {
            props.load(fis);
            wifiFilePath = props.getProperty("WifiFilePath");
            workFilePath = props.getProperty("WorkFilePath");
            isSkipHistory = Integer.parseInt(props.getProperty("IsSkipHistory"));
            isBakFile = Integer.parseInt(props.getProperty("IsBakFile"));
            bakFilePath = props.getProperty("BakFilePath");
            kafkaServer = props.getProperty("KafkaServer");
            isLogPushData = Integer.parseInt(props.getProperty("IsLogPushData"));
            logPushDataPath = props.getProperty("LogPushDataPath");

            logInfo("Configuration loaded successfully");
        } catch (Exception e) {
            logError("Failed to load configuration: " + e.getMessage());
            System.exit(-1);
        }
    }

    /**
     * 初始化H2数据库连接
     */
    private static void initH2Database() {
        try {
            Class.forName("org.h2.Driver");
            String url = "jdbc:h2:./obud_data;AUTO_SERVER=TRUE";
            h2Connection = DriverManager.getConnection(url);
            logInfo("H2 database connected successfully");
        } catch (Exception e) {
            logError("Failed to connect to H2 database: " + e.getMessage());
            System.exit(-1);
        }
    }

    /**
     * 地理坐标转换：GCJ02转BD09
     */
    public static double[] gcj02ToBd09(double lng, double lat) {
        double z = Math.sqrt(lng * lng + lat * lat) + 2e-5 * Math.sin(lat * X_PI);
        double theta = Math.atan2(lat, lng) + 3e-6 * Math.cos(lng * X_PI);
        double bdLng = z * Math.cos(theta) + 0.0065;
        double bdLat = z * Math.sin(theta) + 0.006;
        return new double[]{bdLng, bdLat};
    }

    /**
     * 地理坐标转换：WGS84转GCJ02
     */
    public static double[] wgs84ToGcj02(double lng, double lat) {
        if (outOfChina(lng, lat)) {
            return new double[]{lng, lat};
        }

        double dLat = transformLat(lng - 105.0, lat - 35.0);
        double dLng = transformLng(lng - 105.0, lat - 35.0);

        double radLat = lat / 180.0 * PI;
        double magic = Math.sin(radLat);
        magic = 1 - EE * magic * magic;
        double sqrtMagic = Math.sqrt(magic);

        dLat = dLat * 180.0 / ((A * (1 - EE)) / (magic * sqrtMagic) * PI);
        dLng = dLng * 180.0 / (A / sqrtMagic * Math.cos(radLat) * PI);

        double mgLat = lat + dLat;
        double mgLng = lng + dLng;

        return new double[]{mgLng, mgLat};
    }

    /**
     * 判断是否在中国范围内
     */
    private static boolean outOfChina(double lng, double lat) {
        return !(lng > 73.66 && lng < 135.05 && lat > 3.86 && lat < 53.55);
    }

    /**
     * 转换纬度
     */
    private static double transformLat(double lng, double lat) {
        double ret = -100.0 + 2.0 * lng + 3.0 * lat + 0.2 * lat * lat + 0.1 * lng * lat + 0.2 * Math.sqrt(Math.abs(lng));
        ret += (20.0 * Math.sin(6.0 * lng * PI) + 20.0 * Math.sin(2.0 * lng * PI)) * 2.0 / 3.0;
        ret += (20.0 * Math.sin(lat * PI) + 40.0 * Math.sin(lat / 3.0 * PI)) * 2.0 / 3.0;
        ret += (160.0 * Math.sin(lat / 12.0 * PI) + 320 * Math.sin(lat * PI / 30.0)) * 2.0 / 3.0;
        return ret;
    }

    /**
     * 转换经度
     */
    private static double transformLng(double lng, double lat) {
        double ret = 300.0 + lng + 2.0 * lat + 0.1 * lng * lng + 0.1 * lng * lat + 0.1 * Math.sqrt(Math.abs(lng));
        ret += (20.0 * Math.sin(6.0 * lng * PI) + 20.0 * Math.sin(2.0 * lng * PI)) * 2.0 / 3.0;
        ret += (20.0 * Math.sin(lng * PI) + 40.0 * Math.sin(lng / 3.0 * PI)) * 2.0 / 3.0;
        ret += (150.0 * Math.sin(lng / 12.0 * PI) + 300.0 * Math.sin(lng / 30.0 * PI)) * 2.0 / 3.0;
        return ret;
    }

    /**
     * 获取文件修改时间
     */
    private static long getFileModifyTime(String filePath) {
        try {
            return Files.getLastModifiedTime(Paths.get(filePath)).toMillis();
        } catch (IOException e) {
            logError("Failed to get file modify time: " + e.getMessage());
            return 0;
        }
    }

    /**
     * 获取设备字典
     */
    private static void getDeviceDict() {
        try {
            Map<String, DeviceInfo> newDevDict = new HashMap<>();

            String sql = "SELECT CONCAT(dev_no, station_no) AS ds_no, station_no, id, dev_no, " +
                    "longitude, latitude FROM t_device GROUP BY dev_no, station_no";

            try (PreparedStatement stmt = h2Connection.prepareStatement(sql);
                 ResultSet rs = stmt.executeQuery()) {

                while (rs.next()) {
                    String dsNo = rs.getString("ds_no");
                    String stationNo = rs.getString("station_no");
                    String devNo = rs.getString("dev_no");
                    int id = rs.getInt("id");
                    String longitude = rs.getString("longitude");
                    String latitude = rs.getString("latitude");

                    DeviceInfo deviceInfo = new DeviceInfo(id, stationNo, devNo, longitude, latitude);
                    newDevDict.put(dsNo, deviceInfo);
                }
            }

            devDict.clear();
            devDict.putAll(newDevDict);
            logInfo("Device dictionary loaded successfully: " + devDict.size());
        } catch (SQLException e) {
            logError("Failed to load device dictionary: " + e.getMessage());
            throw new RuntimeException(e);
        }
    }

    private static List<String[]> readCSV(String filename) {
        List<String[]> records = new ArrayList<>();
        try (BufferedReader br = new BufferedReader(new FileReader(filename))) {
            String line;
            while ((line = br.readLine()) != null) {
                String[] values = line.split(",");
                records.add(values);
            }
        } catch (IOException e) {
            logError("Error reading CSV file: " + e.getMessage());
        }
        return records;
    }

    private static void dispose_ST_BASIC_1001(String filename) {
        try {
            List<String[]> fmsg = readCSV(filename);

            if (fmsg.isEmpty()) {
                return;
            }

            int deviceId = 0;
            String[] firstRow = fmsg.get(0);

            if ("12345678".equals(firstRow[0]) || "ASD00001".equals(firstRow[0])) {
                return;
            }

            String ds_no = firstRow[2] + firstRow[0]; // dev_no + station_no
            if (devDict.containsKey(ds_no)) {
                deviceId = devDict.get(ds_no).id;
            }

            for (String[] row : fmsg) {
                double lon = Double.parseDouble(row[9]);
                double lat = Double.parseDouble(row[8]);
                double[] gcj = wgs84ToGcj02(lon, lat);
                double[] bd = gcj02ToBd09(gcj[0], gcj[1]);

                // 先尝试更新记录
                String updateSql;
                if (Double.parseDouble(row[9]) > 0) {
                    updateSql = String.format(
                            "UPDATE t_device SET " +
                                    "addtime=NOW(), dev_mac='%s', cmp_jgdm='%s', hard_ver='%s', soft_ver='%s', dev_ip='%s', " +
                                    "latitude=%f, longitude=%f, latitude_baidu=%f, longitude_baidu=%f, mng_status=0, last_update_time=NOW() " +
                                    "WHERE station_no='%s' AND dev_no='%s'",
                            row[3], row[4], row[5], row[6], row[7],
                            gcj[1], gcj[0], bd[1], bd[0],
                            row[0], row[2]
                    );
                } else {
                    updateSql = String.format(
                            "UPDATE t_device SET " +
                                    "addtime=NOW(), dev_mac='%s', cmp_jgdm='%s', hard_ver='%s', soft_ver='%s', dev_ip='%s', " +
                                    "mng_status=0, last_update_time=NOW() " +
                                    "WHERE station_no='%s' AND dev_no='%s'",
                            row[3], row[4], row[5], row[6], row[7],
                            row[0], row[2]
                    );
                }

                try (Statement statement = h2Connection.createStatement()) {
                    int updatedRows = statement.executeUpdate(updateSql);

                    // 如果没有更新任何行，则插入新记录
                    if (updatedRows == 0) {
                        String insertSql;
                        if (Double.parseDouble(row[9]) > 0) {
                            insertSql = String.format(
                                    "INSERT INTO t_device (station_no, addtime, dev_no, dev_mac, cmp_jgdm, hard_ver, soft_ver, dev_ip, latitude, longitude, latitude_baidu, longitude_baidu, mng_status) " +
                                            "VALUES ('%s', NOW(), '%s', '%s', '%s', '%s', '%s', '%s', %f, %f, %f, %f, 0)",
                                    row[0], row[2], row[3], row[4], row[5], row[6], row[7],
                                    gcj[1], gcj[0], bd[1], bd[0]
                            );
                        } else {
                            insertSql = String.format(
                                    "INSERT INTO t_device (station_no, addtime, dev_no, dev_mac, cmp_jgdm, hard_ver, soft_ver, dev_ip, latitude, longitude, latitude_baidu, longitude_baidu, mng_status) " +
                                            "VALUES ('%s', NOW(), '%s', '%s', '%s', '%s', '%s', '%s', %f, %f, %f, %f, 0)",
                                    row[0], row[2], row[3], row[4], row[5], row[6], row[7],
                                    gcj[1], gcj[0], bd[1], bd[0]
                            );
                        }
                        statement.executeUpdate(insertSql);
                    }
                } catch (SQLException e) {
                    logError("SQL execution error: " + e.getMessage());
                }
            }

            if (deviceId == 0) {
                getDeviceDict();
            }
        } catch (Exception e) {
            logError("Error processing ST_BASIC_1001: " + e.getMessage());
        }
    }

    private static void dispose_ST_STATUS_1001(String filename) {
        try (Statement statement = h2Connection.createStatement()) {
            List<String[]> fmsg = readCSV(filename);

            if (fmsg.isEmpty()) {
                return;
            }

            String[] firstRow = fmsg.get(0);
            if (firstRow.length < 12) {
                List<String[]> extendedFmsg = new ArrayList<>();
                for (String[] row : fmsg) {
                    String[] newRow = new String[12];
                    System.arraycopy(row, 0, newRow, 0, row.length);
                    for (int i = row.length; i < 12; i++) {
                        newRow[i] = "0";
                    }
                    extendedFmsg.add(newRow);
                }
                fmsg = extendedFmsg;
            }

            for (String[] row : fmsg) {
                if (row[9].isEmpty()) {
                    row[9] = "0";
                    row[10] = "0";
                    row[11] = "0";
                }

                double lon = Double.parseDouble(row[5]);
                double lat = Double.parseDouble(row[4]);
                double[] gcj = wgs84ToGcj02(lon, lat);
                double[] bd = gcj02ToBd09(gcj[0], gcj[1]);

                String sql;
                if (lon > 0) {
                    sql = String.format("UPDATE t_device SET last_update_time=NOW(), longitude=%f, latitude=%f, " +
                                    "longitude_baidu=%f, latitude_baidu=%f WHERE station_no='%s' AND dev_no='%s'",
                            gcj[0], gcj[1], bd[0], bd[1], row[0], row[2]);
                } else {
                    sql = String.format("UPDATE t_device SET last_update_time=NOW() WHERE station_no='%s'" +
                            " AND dev_no='%s'", row[0], row[2]);
                }

                try {
                    statement.executeUpdate(sql);
                } catch (SQLException e) {
                    logError("SQL execution error: " + e.getMessage());
                    try (Statement newCursor = h2Connection.createStatement()) {
                        newCursor.executeUpdate(sql);
                    }
                }

                if ("12345678".equals(firstRow[0]) || "ASD00001".equals(firstRow[0])) {
                    return;
                }

                String ds_no = firstRow[2] + firstRow[0]; // dev_no + station_no
                if (!devDict.containsKey(ds_no)) {
                    String insertSql = String.format("INSERT INTO t_device(station_no,addtime,dev_no,dev_mac,latitude," +
                                    "longitude,latitude_baidu,longitude_baidu,mng_status,run_status, last_update_time)" +
                                    " VALUES ('%s', NOW(), '%s', '%s', '%s', '%s', %f, %f, 0, 1, NOW())",
                            row[0], row[2], row[3], row[4], row[5], bd[1], bd[0]);

                    statement.executeUpdate(insertSql);
                    getDeviceDict();
                }
            }
        } catch (Exception e) {
            logError("Error processing ST_STATUS_1001");
        }
    }

    /**
     * 线程处理ST文件
     */
    private static void threadST() {
        String dealFile = "";

        try {
            while (true) {
                if (threadExit == -1) {
                    return;
                }

                try {
                    List<String> files = getSourceFile(wifiFilePath);
                    Thread.sleep(2000);

                    if (files.isEmpty()) {
                        Thread.sleep(2000);
                    } else {
                        long newFileTime = System.currentTimeMillis() / 1000;

                        for (String file : files) {
                            dealFile = file;

                            long changeTime = getFileModifyTime(file) / 1000;
                            if (newFileTime - changeTime <= 2) {
                                continue;
                            }

                            if (file.endsWith(".zip") || file.endsWith(".zip.tea")) {
                                continue;
                            }

                            if (file.endsWith(".ST_BASIC_1001")) {
                                dispose_ST_BASIC_1001(file);
                            } else if (file.endsWith(".ST_STATUS_1001")) {
                                dispose_ST_STATUS_1001(file);
                            }

                            try {
                                Files.move(Paths.get(file), Paths.get(workFilePath + "/" + Paths.get(file).getFileName()));
                            } catch (Exception e) {
                                Files.deleteIfExists(Paths.get(file));
                            }
                        }
                    }
                } catch (Exception e) {
                    logError("Error in threadST: " + e.getMessage());
                    try {
                        Files.deleteIfExists(Paths.get(dealFile));
                    } catch (Exception ex) {
                        logError("Failed to delete file: " + ex.getMessage());
                    }
                }
            }
        } catch (Exception e) {
            logError("Error in threadST: " + e.getMessage());
        }
    }

    /**
     * 获取数据文件
     */
    private static List<String> getSourceFile(String dir) {
        List<String> allFiles = new ArrayList<>();

        try (Stream<Path> pathStream = Files.walk(Paths.get(dir))) {
            pathStream.filter(path -> {
                // 首先检查文件是否存在，避免NoSuchFileException
                if (!Files.exists(path)) {
                    return false;
                }

                // 检查是否为常规文件
                if (!Files.isRegularFile(path)) {
                    return false;
                }

                String fileName = path.getFileName().toString();
                if (fileName.contains("testcfg") || fileName.contains(".zip")) {
                    return false;
                }

                return fileName.contains(".ST_");
            }).limit(100).forEach(path -> allFiles.add(path.toString()));
        } catch (UncheckedIOException ignored) {
        } catch (Exception e) {
            logError("Error getting files from directory: " + e.getMessage());
        }

        return allFiles;
    }

    /**
     * 创建设备表
     */
    private static void createDeviceTable() {
        try {
            String sql = "CREATE TABLE IF NOT EXISTS t_device (" +
                    "id INT AUTO_INCREMENT PRIMARY KEY, " +
                    "site_no VARCHAR(32), " +
                    "dev_no VARCHAR(32) NOT NULL, " +
                    "station_no VARCHAR(12) NOT NULL, " +
                    "dev_name VARCHAR(50), " +
                    "dev_mac VARCHAR(12) NOT NULL, " +
                    "dev_location VARCHAR(128), " +
                    "location_type TINYINT, " +
                    "addtime TIMESTAMP, " +
                    "user_id INT, " +
                    "city_code VARCHAR(16), " +
                    "county_code VARCHAR(16), " +
                    "longitude DOUBLE, " +
                    "latitude DOUBLE, " +
                    "orientation TINYINT, " +
                    "dev_ip VARCHAR(15), " +
                    "last_update_time TIMESTAMP, " +
                    "run_status SMALLINT, " +
                    "mng_status TINYINT, " +
                    "net_code INT DEFAULT 0, " +
                    "dev_model VARCHAR(10), " +
                    "dev_level TINYINT DEFAULT 0, " +
                    "dev_sim VARCHAR(15), " +
                    "dev_type TINYINT, " +
                    "soft_ver VARCHAR(32), " +
                    "hard_ver VARCHAR(32) DEFAULT 'GN-W-V2.0', " +
                    "cmp_jgdm VARCHAR(10), " +
                    "up_interval INT, " +
                    "scan_radius INT, " +
                    "town_code VARCHAR(16), " +
                    "city_name VARCHAR(32), " +
                    "county_name VARCHAR(32), " +
                    "town_name VARCHAR(32), " +
                    "net_service_provider VARCHAR(8), " +
                    "site_id VARCHAR(64), " +
                    "device_id VARCHAR(64), " +
                    "longitude_original DOUBLE, " +
                    "latitude_original DOUBLE, " +
                    "longitude_baidu DOUBLE, " +
                    "latitude_baidu DOUBLE, " +
                    "dev_period TINYINT, " +
                    "new_dev_type INT, " +
                    "service_status TINYINT, " +
                    "hp_xn INT DEFAULT 0, " +
                    "CONSTRAINT dno UNIQUE (dev_no, station_no))";

            try (Statement stmt = h2Connection.createStatement()) {
                stmt.execute(sql);
            }

            logInfo("Device tables created successfully");
        } catch (SQLException e) {
            logError("Failed to create device tables: " + e.getMessage());
        }
    }

    public static void unzipThread() {
        while (true) {
            try (Stream<Path> pathStream = Files.list(Paths.get(wifiFilePath))) {
                pathStream.filter(Files::isRegularFile).filter(path -> path.toString().endsWith(".zip")).forEach(path -> {
                    try {
                        try (ZipFile zipFile = new ZipFile(path.toString())) {
                            zipFile.stream().forEach(entry -> {
                                try {
                                    Path extractedPath = Paths.get(wifiFilePath, entry.getName());
                                    Files.copy(zipFile.getInputStream(entry), extractedPath,
                                            StandardCopyOption.REPLACE_EXISTING);
                                } catch (IOException e) {
                                    logError("解压文件失败: " + e.getMessage());
                                }
                            });
                        }
                        Files.deleteIfExists(path);
                    } catch (IOException e) {
                        logError("解压文件错误: " + e.getMessage());

                        try {
                            Files.deleteIfExists(path);
                        } catch (IOException ex) {
                            logError("删除压缩文件失败: " + ex.getMessage());
                        }
                    }
                });

                Thread.sleep(2000);
            } catch (IOException | InterruptedException e) {
                logError("解压线程异常: " + e.getMessage());
            }
        }
    }

    /**
     * 解密TEA文件
     */
    private static void teaFileDecrypt() {
        File inputDirectory = new File(wifiFilePath);
        File outputDirectory = new File(wifiFilePath);

        if (!outputDirectory.exists()) {
            String outputDirErr = "输出目录不存在: " + wifiFilePath;
            System.err.println(outputDirErr);
            throw new RuntimeException(outputDirErr);
        }

        while (true) {
            try {
                TimeUnit.SECONDS.sleep(1L);
                File[] files = inputDirectory.listFiles();
                if (files != null) {
                    for (File file : files) {
                        if (!file.isHidden() && !file.isDirectory() && isTeaFile(file)) {
                            processTeaFile(file, wifiFilePath);
                        }
                    }
                }
            } catch (InterruptedException var11) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                System.err.println("处理文件时发生错误: " + e.getMessage());
            }
        }
    }

    private static boolean isTeaFile(File file) {
        String fileName = file.getName();
        if (file.getName().startsWith(".") || !fileName.toLowerCase().endsWith(".zip.tea")) {
            return false;
        } else {
            long fileTime = file.lastModified();
            long currentTime = System.currentTimeMillis();
            long timeDiff = (currentTime - fileTime) / 1000L;
            return timeDiff > 2L;
        }
    }

    private static void processTeaFile(File inputFile, String outputDir) {
        try {
            String outputFileName = inputFile.getName().substring(0, inputFile.getName().length() - 4);
            File outputFile = new File(outputDir, outputFileName);
            decryptTEA(inputFile, outputFile);
            Files.delete(inputFile.toPath());
        } catch (IOException e) {
            logError("处理文件时出错: " + inputFile);
            System.err.println("处理文件时出错: " + e.getMessage());
        }
    }

    private static void decryptTEA(File input, File output) {
        try (FileInputStream inputFile = new FileInputStream(input)) {
            FileOutputStream outputFile = new FileOutputStream(output);
            Throwable var5 = null;

            try {
                byte[] buffer = new byte[8];

                int bytesRead;
                while ((bytesRead = inputFile.read(buffer)) == 8) {
                    int[] block = bytesToInts(buffer);
                    teaDecrypt(block, TEA_KEY);
                    byte[] decryptedBytes = intsToBytes(block);
                    outputFile.write(decryptedBytes);
                }

                if (bytesRead > 0 && bytesRead < 8) {
                    for (int i = bytesRead; i < 8; ++i) {
                        buffer[i] = 0;
                    }

                    int[] block = bytesToInts(buffer);
                    teaDecrypt(block, TEA_KEY);
                    byte[] decryptedBytes = intsToBytes(block);
                    outputFile.write(decryptedBytes);
                }
            } catch (Throwable var33) {
                var5 = var33;
                throw var33;
            } finally {
                if (var5 != null) {
                    try {
                        outputFile.close();
                    } catch (Throwable var32) {
                        var5.addSuppressed(var32);
                    }
                } else {
                    outputFile.close();
                }

            }
        } catch (IOException e) {
            logError("Error reading from input file: " + e.getMessage());
        }

    }

    private static byte[] intsToBytes(int[] ints) {
        if (ints.length != 2) {
            throw new IllegalArgumentException("Input int array must be exactly 2 integers");
        } else {
            ByteBuffer buffer = ByteBuffer.allocate(8);
            buffer.order(ByteOrder.LITTLE_ENDIAN);
            buffer.putInt(ints[0]);
            buffer.putInt(ints[1]);
            return buffer.array();
        }
    }

    public static void teaDecrypt(int[] data, int[] key) {
        int uVar1 = Integer.reverseBytes(data[0]);
        int uVar2 = Integer.reverseBytes(data[1]);
        int iVar3 = -957401312;

        do {
            uVar2 -= (uVar1 << 4) + key[2] ^ (uVar1 >>> 5) + key[3] ^ iVar3 + uVar1;
            uVar1 -= (uVar2 << 4) + key[0] ^ (uVar2 >>> 5) + key[1] ^ uVar2 + iVar3;
            iVar3 += 1640531527;
        } while (iVar3 != 0);

        data[0] = Integer.reverseBytes(uVar1);
        data[1] = Integer.reverseBytes(uVar2);
    }

    private static int[] bytesToInts(byte[] bytes) {
        if (bytes.length != 8) {
            throw new IllegalArgumentException("Input byte array must be exactly 8 bytes");
        } else {
            int[] result = new int[2];
            ByteBuffer buffer = ByteBuffer.wrap(bytes);
            buffer.order(ByteOrder.LITTLE_ENDIAN);
            result[0] = buffer.getInt();
            result[1] = buffer.getInt();
            return result;
        }
    }

    private static KafkaProducer<String, String> createKafkaProducer() {
        Properties props = new Properties();

        // 基础配置
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 性能配置
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384 * 4);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 2000);
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000);
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5000);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        // 安全认证配置
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required " +
                "username=\"muju\" " + "password=\"muju@2022\";");

        return new KafkaProducer<>(props);
    }

    /**
     * 同步设备信息到 Kafka
     */
    private static void syncDeviceToKafka() {
        try (KafkaProducer<String, String> producer = createKafkaProducer()) {
            try (PreparedStatement ps = h2Connection.prepareStatement(
                    "SELECT station_no as Station_no, dev_no as Device_no, dev_name, run_status, " +
                            "longitude, latitude, EXTRACT(EPOCH FROM last_update_time) as update_time, " +
                            "dev_ip as ip FROM t_device WHERE mng_status=1 AND dev_name IS NOT NULL")) {
                ResultSet rs = ps.executeQuery();
                while (rs.next()) {
                    JsonObject obj = new JsonObject();
                    obj.addProperty("Station_no", rs.getString("Station_no"));
                    obj.addProperty("Device_no", rs.getString("Device_no"));
                    obj.addProperty("dev_name", rs.getString("dev_name"));
                    obj.addProperty("run_status", rs.getInt("run_status"));
                    obj.addProperty("longitude", rs.getString("longitude"));
                    obj.addProperty("latitude", rs.getString("latitude"));
                    obj.addProperty("update_time", rs.getLong("update_time"));
                    obj.addProperty("ip", rs.getString("ip"));

                    producer.send(new ProducerRecord<>("device_details", obj.toString()));
                }
            }

            logInfo("同步设备信息到 Kafka");
        } catch (Exception e) {
            logError("设备信息同步失败" + e.getMessage());
        }
    }

    // 获取指定后缀的文件列表
    private static List<File> getFilesWithExt(String dirPath, String ext) {
        List<File> files = new ArrayList<>();
        File dir = new File(dirPath);
        if (!dir.exists() || !dir.isDirectory()) return files;

        File[] allFiles = dir.listFiles((d, name) -> name.toLowerCase().endsWith("." + ext.toLowerCase()));
        if (allFiles != null) {
            files.addAll(Arrays.asList(allFiles));
        }
        return files;
    }

    private static class FileProcessorThread implements Runnable {
        private final String fileType;
        private final BiConsumer<File, KafkaProducer<String, String>> processor;

        public FileProcessorThread(String fileType, BiConsumer<File, KafkaProducer<String, String>> processor) {
            this.fileType = fileType;
            this.processor = processor;
        }

        @Override
        public void run() {
            KafkaProducer<String, String> kafkaProducer = createKafkaProducer();
            int errorCount = 0;

            while (true) {
                if (threadExit == -1) {
                    break;
                }

                try {
                    List<File> files = getFilesWithExt(workFilePath, fileType);
                    for (File file : files) {
                        if (System.currentTimeMillis() - file.lastModified() < 2000) continue;

                        if (processor != null) {
                            processor.accept(file, kafkaProducer);
                        }

                        // 移动或删除文件
                        if (isBakFile == 1) {
                            File bakDir = new File(bakFilePath);
                            bakDir.mkdirs();
                            File newFile = new File(bakDir, file.getName());
                            file.renameTo(newFile);
                        } else {
                            file.delete();
                        }
                    }
                } catch (Exception e) {
                    logError("文件处理线程异常: " + e.getMessage());
                    errorCount++;

                    // 如果错误过多，重新创建producer
                    if (errorCount % 10 == 1) {
                        try {
                            kafkaProducer.close();
                        } catch (Exception closeEx) {
                            logError("关闭旧producer失败: " + closeEx.getMessage());
                        }
                        kafkaProducer = createKafkaProducer();
                    }
                }

                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }

            // 确保在退出时关闭producer
            try {
                kafkaProducer.close();
            } catch (Exception e) {
                logError("关闭producer时出错: " + e.getMessage());
            }
        }
    }

    /**
     * 发送基础信息到Kafka
     */
    private static void producerSendJsonBasic(File file, KafkaProducer<String, String> producer) {
        try {
            List<Map<String, Object>> records = readCsvFile(file.getPath(), Arrays.asList(
                    "Station_no", "capture_time", "Device_no", "dev_mac", "factory_code",
                    "firmware", "soft_version", "dev_ip", "Latitude", "Longitude"
            ), Arrays.asList("Station_no", "Device_no", "dev_mac", "factory_code",
                    "firmware", "soft_version", "dev_ip"));

            if (!records.isEmpty()) {
                String topic = "YN_BASIC_1001";
                for (Map<String, Object> record : records) {
                    try {
                        String msg = mapToJson(record);
                        producer.send(new ProducerRecord<>(topic, msg));
                    } catch (Exception e) {
                        logError("Error sending message: " + e.getMessage());
                        errorNo.incrementAndGet();
                    }
                }
            }
        } catch (Exception e) {
            logError("Error in Producer_Send_Json_Basic: " + e.getMessage());
        }
    }

    /**
     * 发送状态信息到Kafka
     */
    private static void producerSendJsonStatus(File file, KafkaProducer<String, String> producer) {
        try {
            List<Map<String, Object>> records = readCsvFile(file.getPath(), Arrays.asList(
                    "Station_no", "capture_time", "Device_no", "dev_mac", "Latitude",
                    "Longitude", "netflow", "cpuload", "runtime", "power", "chl", "track"
            ), Arrays.asList("Station_no", "Device_no", "dev_mac"));

            if (!records.isEmpty()) {
                List<String> messageBuffer = new ArrayList<>();

                // 处理CPU负载
                for (Map<String, Object> record : records) {
                    Object cpuload = record.get("cpuload");
                    if (cpuload != null) {
                        try {
                            double load = Double.parseDouble(cpuload.toString());
                            record.put("cpuload", load * 100);
                        } catch (NumberFormatException e) {
                            record.put("cpuload", 0);
                        }
                    }
                }

                String stationNo = records.get(0).get("Station_no").toString();
                String deviceNo = records.get(0).get("Device_no").toString();
                String key = stationNo + "_" + deviceNo;

                double longitude = 0;
                double latitude = 0;
                DeviceInfo deviceInfo = devDict.get(key);
                if (deviceInfo != null) {
                    longitude = Double.parseDouble(deviceInfo.longitude);
                    latitude = Double.parseDouble(deviceInfo.latitude);
                }

                // 替换经纬度
                for (Map<String, Object> record : records) {
                    if (record.get("Longitude").toString().equals("0")) {
                        record.put("Longitude", longitude);
                    }
                    if (record.get("Latitude").toString().equals("0")) {
                        record.put("Latitude", latitude);
                    }
                }

                String topic = "YN_STATUS_1001";
                for (Map<String, Object> record : records) {
                    try {
                        String msg = mapToJson(record);
                        producer.send(new ProducerRecord<>(topic, msg));

                        if (isLogPushData == 1) {
                            messageBuffer.add(msg);
                        }
                    } catch (Exception e) {
                        logError("Error sending message: " + e.getMessage());
                    }
                }

                if (isLogPushData == 1 && !messageBuffer.isEmpty()) {
                    String topic1 = String.format("QCFK_ETC_ST-YN_STATUS_1001_%d.json", Instant.now().getEpochSecond());
                    writeToFile(messageBuffer, topic1);
                }
            }
        } catch (Exception e) {
            logError("Error in Producer_Send_Json_Status: " + e.getMessage());
        }
    }

    /**
     * 发送RD_1001数据到Kafka
     */
    private static void producerSendJsonRD1001(File file, KafkaProducer<String, String> producer) {
        try {
            List<Map<String, Object>> records = readCsvFile(file.getPath(), Arrays.asList(
                    "Station_no", "Device_no", "PassCardID", "obu", "capture_time", "EtcRegPlateNum", "EtcRegColor",
                    "EtcRegType", "usr_type", "operate", "times", "BillingCardID", "card_type", "card_ver",
                    "card_start", "card_end"
            ), Arrays.asList("Station_no", "Device_no", "PassCardID", "obu", "BillingCardID"));

            String fileName = file.getPath();
            // 尝试读取1002文件
            String file2 = fileName.substring(0, fileName.length() - 4) + "1002";
            List<Map<String, Object>> records2;
            if (Files.exists(Paths.get(file2))) {
                records2 = readCsvFile(file2, Arrays.asList(
                        "Station_no", "Device_no", "PassCardID", "capture_time", "EtcRegPlateNum", "EtcRegName",
                        "EtcRegIDNumber", "card_type", "EtcRegType", "EtcRegColor"
                ), Arrays.asList("Station_no", "Device_no", "PassCardID", "EtcRegName", "EtcRegIDNumber"));
            } else {
                // 创建空的DataFrame结构
                records2 = new ArrayList<>();
            }

            // 合并数据
            Map<String, Map<String, Object>> records2Map = records2.stream().collect(Collectors.toMap(record ->
                    record.get("PassCardID").toString(), Function.identity(), (existing, replacement) -> existing));

            for (Map<String, Object> record : records) {
                String passCardId = record.get("PassCardID").toString();
                Map<String, Object> additionalData = records2Map.get(passCardId);
                if (additionalData != null) {
                    record.put("EtcRegName", additionalData.get("EtcRegName"));
                    record.put("EtcRegIDNumber", additionalData.get("EtcRegIDNumber"));
                }
            }

            records = captureTimeFilter(records);

            if (!records.isEmpty()) {
                String stationNo = records.get(0).get("Station_no").toString();
                String deviceNo = records.get(0).get("Device_no").toString();
                String key = stationNo + "_" + deviceNo;

                double longitude = 0;
                double latitude = 0;
                DeviceInfo deviceInfo = devDict.get(key);
                if (deviceInfo != null) {
                    longitude = Double.parseDouble(deviceInfo.longitude);
                    latitude = Double.parseDouble(deviceInfo.latitude);
                }

                List<String> messageBuffer = new ArrayList<>();

                for (Map<String, Object> record : records) {
                    // 处理PassCardID
                    String passCardId = record.get("PassCardID").toString();
                    String artPassCardId = passCardId.compareTo("a0000000") >= 0 ? passCardId : "";
                    String newPassCardId = passCardId.compareTo("a0000000") >= 0 ? "" : passCardId;
                    record.put("ArtPassCardID", artPassCardId);
                    record.put("PassCardID", newPassCardId);

                    // 解码车牌和姓名
                    record.put("EtcRegPlateNum", mydecode(record.get("EtcRegPlateNum")));
                    record.put("EtcRegName", mydecode(record.get("EtcRegName")));

                    // GB18030解码
                    record.put("EtcRegAddr", hexToGb18030(record.get("obu").toString()));

                    // 设置经纬度
                    record.put("Longitude", longitude);
                    record.put("Latitude", latitude);

                    // 移除不需要的字段
                    record.remove("usr_type");
                    record.remove("times");
                    record.remove("card_type");
                    record.remove("card_ver");
                    record.remove("card_start");
                    record.remove("card_end");
                    record.remove("obu");
                    record.remove("operate");

                    try {
                        String msg = mapToJson(record);
                        producer.send(new ProducerRecord<>("YN_SOURCE_RD_1001", msg));

                        if (isLogPushData == 1) {
                            messageBuffer.add(msg);
                        }
                    } catch (Exception e) {
                        errorNo.incrementAndGet();
                        logError("ProducerRecord Error: " + e.getMessage());
                        return;
                    }
                }

                if (isLogPushData == 1 && !messageBuffer.isEmpty()) {
                    String topic1 = String.format("QCFK_ETC-YN_SOURCE_RD_1001_%d.json", Instant.now().getEpochSecond());
                    writeToFile(messageBuffer, topic1);
                }
            }
        } catch (Exception e) {
            logError("Error in Producer_Send_Json_RD_1001: " + e.getMessage());
        }
    }

    /**
     * 发送RD_1002数据到Kafka
     */
    private static void producerSendJsonRD1002(File file, KafkaProducer<String, String> producer) {
        try {
            List<Map<String, Object>> records = readCsvFile(file.getPath(), Arrays.asList(
                    "Station_no", "Device_no", "PassCardID", "capture_time", "EtcRegPlateNum", "EtcRegName",
                    "EtcRegIDNumber", "card_type", "EtcRegType", "EtcRegColor"
            ), Arrays.asList("Station_no", "Device_no", "PassCardID", "EtcRegName", "EtcRegIDNumber"));

            records = captureTimeFilter(records);

            if (!records.isEmpty()) {
                String stationNo = records.get(0).get("Station_no").toString();
                String deviceNo = records.get(0).get("Device_no").toString();
                String key = stationNo + "_" + deviceNo;

                double longitude = 0;
                double latitude = 0;
                DeviceInfo deviceInfo = devDict.get(key);
                if (deviceInfo != null) {
                    longitude = Double.parseDouble(deviceInfo.longitude);
                    latitude = Double.parseDouble(deviceInfo.latitude);
                }

                List<String> messageBuffer = new ArrayList<>();

                for (Map<String, Object> record : records) {
                    // 解码车牌和姓名
                    record.put("EtcRegName", mydecode(record.get("EtcRegName")));
                    record.put("EtcRegPlateNum", mydecode(record.get("EtcRegPlateNum")));

                    // 设置固定值
                    record.put("ArtPassCardID", "");
                    record.put("BillingCardID", "");
                    record.put("EtcRegAddr", "");
                    record.put("Longitude", longitude);
                    record.put("Latitude", latitude);

                    try {
                        String msg = mapToJson(record);
                        producer.send(new ProducerRecord<>("YN_SOURCE_RD_1001", msg));

                        if (isLogPushData == 1) {
                            messageBuffer.add(msg);
                        }
                    } catch (Exception e) {
                        logError(e.getMessage());
                    }
                }

                if (isLogPushData == 1 && !messageBuffer.isEmpty()) {
                    String topic1 = String.format("QCFK_ETC-YN_SOURCE_RD_1001_%d.json", Instant.now().getEpochSecond());
                    writeToFile(messageBuffer, topic1);
                }
            }
        } catch (Exception e) {
            logError("Error in producerSendJsonRD1002: " + e.getMessage());
        }
    }

    /**
     * 发送RD_1004数据到Kafka
     */
    private static void producerSendJsonRD1004(File file, KafkaProducer<String, String> producer) {
        try {
            List<Map<String, Object>> records = readCsvFile(file.getPath(), Arrays.asList(
                    "Station_no", "Device_no", "PassCardID", "capture_time",
                    "EtcRegPlateNum", "road_no", "cash_no", "drive_time",
                    "drive_status", "car_type", "color"
            ), Arrays.asList("Station_no", "Device_no", "PassCardID", "road_no", "cash_no"));

            captureTimeFilter(records);

            if (!records.isEmpty()) {
                String stationNo = records.get(0).get("Station_no").toString();
                String deviceNo = records.get(0).get("Device_no").toString();
                String key = stationNo + "_" + deviceNo;

                double longitude = 0;
                double latitude = 0;
                DeviceInfo deviceInfo = devDict.get(key);
                if (deviceInfo != null) {
                    longitude = Double.parseDouble(deviceInfo.longitude);
                    latitude = Double.parseDouble(deviceInfo.latitude);
                }

                List<String> messageBuffer = new ArrayList<>();

                for (Map<String, Object> record : records) {
                    // 解码车牌
                    record.put("EtcRegPlateNum", mydecode(record.get("EtcRegPlateNum")));

                    // 设置经纬度
                    record.put("Longitude", longitude);
                    record.put("Latitude", latitude);

                    // 跳过空车牌
                    if (" ".equals(record.get("EtcRegPlateNum").toString())) {
                        continue;
                    }

                    try {
                        String msg = mapToJson(record);
                        producer.send(new ProducerRecord<>("YN_SOURCE_RD_1004", msg));

                        logInfo("YN_SOURCE_RD_1004 : " + msg);

                        if (isLogPushData == 1) {
                            messageBuffer.add(msg);
                        }
                    } catch (Exception e) {
                        logError("Error sending message: " + e.getMessage());
                        errorNo.incrementAndGet();
                    }
                }

                if (isLogPushData == 1 && !messageBuffer.isEmpty()) {
                    String topic1 = String.format("QCFK_ETC-YN_SOURCE_RD_1004_%d.json", Instant.now().getEpochSecond());
                    writeToFile(messageBuffer, topic1);
                }
            }
        } catch (Exception e) {
            logError("Error in Producer_Send_Json_RD_1004: " + e.getMessage());
        }
    }

    /**
     * 发送RD_1005数据到Kafka
     */
    private static void producerSendJsonRD1005(File file, KafkaProducer<String, String> producer) {
        try {
            List<Map<String, Object>> records = readCsvFile(file.getPath(), Arrays.asList(
                    "Station_no", "Device_no", "PassCardID", "capture_time",
                    "EtcRegPlateNum", "road_no", "cash_no", "drive_time",
                    "drive_status", "road_no_out", "cash_no_out", "drive_time_out",
                    "drive_status_out", "car_type", "color", "usr_type", "pro_cnt",
                    "amount_rec", "amount_paid", "paid_cnt", "mile_acc", "Longitude", "Latitude"
            ), Arrays.asList("Station_no", "Device_no", "PassCardID", "EtcRegPlateNum",
                    "road_no", "cash_no", "road_no_out", "cash_no_out"));

            if (records.isEmpty()) {
                return;
            }

            records = captureTimeFilter(records);

            if (records.isEmpty()) {
                return;
            }

            // 过滤不同的道路和收费站
            records = records.stream().filter(record -> !record.get("road_no").equals(record.get("road_no_out"))
                    || !record.get("cash_no").equals(record.get("cash_no_out"))).collect(Collectors.toList());

            // 过滤里程大于0
            records = records.stream().filter(record -> {
                try {
                    int mileAcc = Integer.parseInt(record.get("mile_acc").toString());
                    return mileAcc > 0;
                } catch (NumberFormatException e) {
                    return false;
                }
            }).collect(Collectors.toList());

            if (records.isEmpty()) {
                return;
            }

            Map<String, Object> firstRecord = records.get(0);
            String stationNo = firstRecord.get("Station_no").toString();
            String deviceNo = firstRecord.get("Device_no").toString();
            String key = stationNo + "_" + deviceNo;

            double longitude = 0;
            double latitude = 0;
            DeviceInfo deviceInfo = devDict.get(key);
            if (deviceInfo != null) {
                longitude = Double.parseDouble(deviceInfo.longitude);
                latitude = Double.parseDouble(deviceInfo.latitude);
            }

            List<String> messageBuffer = new ArrayList<>();

            for (Map<String, Object> record : records) {
                // 解码车牌
                record.put("EtcRegPlateNum", mydecode(record.get("EtcRegPlateNum")));

                // 设置经纬度
                record.put("Longitude", longitude);
                record.put("Latitude", latitude);

                // 跳过空车牌
                if (" ".equals(record.get("EtcRegPlateNum").toString())) {
                    continue;
                }

                try {
                    String msg = mapToJson(record);
                    producer.send(new ProducerRecord<>("YN_SOURCE_RD_1005", msg));

                    if (isLogPushData == 1) {
                        messageBuffer.add(msg);
                    }
                } catch (Exception e) {
                    logError("Error sending message: " + e.getMessage());
                    errorNo.incrementAndGet();
                }
            }

            if (isLogPushData == 1 && !messageBuffer.isEmpty()) {
                String topic1 = String.format("QCFK_ETC-YN_SOURCE_RD_1005_%d.json", Instant.now().getEpochSecond());
                writeToFile(messageBuffer, topic1);
            }
        } catch (Exception e) {
            logError("Error in Producer_Send_Json_RD_1005: " + e.getMessage());
        }
    }

    private static List<Map<String, Object>> captureTimeFilter(List<Map<String, Object>> records) {
        if (isSkipHistory == 0) {
            return records;
        }

        long stime = System.currentTimeMillis() - 86400 * 7;
        long etime = System.currentTimeMillis() + 86400;

        // 过滤时间范围
        records = records.stream().filter(record -> {
            try {
                long captureTime = Long.parseLong(record.get("capture_time").toString());
                return captureTime < etime && captureTime > stime;
            } catch (NumberFormatException e) {
                return false;
            }
        }).collect(Collectors.toList());
        return records;
    }

    private static List<Map<String, Object>> readCsvFile(String filename, List<String> columnNames, List<String> stringColumns) {
        List<Map<String, Object>> records = new ArrayList<>();
        try (BufferedReader br = new BufferedReader(new FileReader(filename))) {
            String line;
            while ((line = br.readLine()) != null) {
                String[] values = line.split(",");
                Map<String, Object> record = new HashMap<>();

                for (int i = 0; i < Math.min(columnNames.size(), values.length); i++) {
                    String columnName = columnNames.get(i);
                    String value = values[i].trim();

                    if (stringColumns.contains(columnName)) {
                        record.put(columnName, value.isEmpty() ? "0" : value);
                    } else {
                        // 尝试转换为数字，如果失败则保留字符串
                        try {
                            if (value.contains(".")) {
                                record.put(columnName, Double.parseDouble(value));
                            } else {
                                record.put(columnName, Long.parseLong(value));
                            }
                        } catch (NumberFormatException e) {
                            record.put(columnName, value.isEmpty() ? 0 : value);
                        }
                    }
                }

                records.add(record);
            }
        } catch (IOException e) {
            logError("Error reading CSV file: " + e.getMessage());
        }

        return records;
    }

    private static String mapToJson(Map<String, Object> map) {
        StringBuilder json = new StringBuilder("{");
        boolean first = true;

        for (Map.Entry<String, Object> entry : map.entrySet()) {
            if (!first) {
                json.append(",");
            }
            json.append("\"").append(entry.getKey()).append("\":");

            Object value = entry.getValue();
            if (value instanceof String) {
                json.append("\"").append(escapeJson(value.toString())).append("\"");
            } else {
                json.append(value);
            }

            first = false;
        }

        json.append("}");
        return json.toString();
    }

    private static String escapeJson(String input) {
        return input.replace("\\", "\\\\")
                .replace("\"", "\\\"")
                .replace("\n", "\\n")
                .replace("\r", "\\r")
                .replace("\t", "\\t");
    }

    private static String hexToGb18030(String hexString) {
        try {
            // 简化实现，实际需要GB18030编码支持
            if (hexString == null || hexString.isEmpty()) {
                return "";
            }

            // 移除空格和0x前缀
            hexString = hexString.replaceAll("\\s+", "").replaceAll("^(0x|0X)", "");

            byte[] bytes = new byte[hexString.length() / 2];
            for (int i = 0; i < bytes.length; i++) {
                int index = i * 2;
                int v = Integer.parseInt(hexString.substring(index, index + 2), 16);
                bytes[i] = (byte) v;
            }

            String result = new String(bytes, "GBK"); // 使用GBK作为近似
            return result.replace("\0", "");
        } catch (Exception e) {
            return "";
        }
    }

    private static String mydecode(Object code) {
        try {
            if (code == null) {
                return " ";
            }
            String strCode = code.toString();
            // 简化实现，实际需要GB18030解码
            return new String(strCode.getBytes(StandardCharsets.ISO_8859_1), "GBK").replace("?", "");
        } catch (Exception e) {
            return " ";
        }
    }

    private static void writeToFile(List<String> messageBuffer, String filename) {
        try (BufferedWriter writer = Files.newBufferedWriter(Paths.get(logPushDataPath, filename))) {
            for (int i = 0; i < messageBuffer.size(); i++) {
                writer.write(messageBuffer.get(i));
                if (i < messageBuffer.size() - 1) {
                    writer.newLine();
                }
            }
        } catch (IOException e) {
            logError("Error writing to file: " + e.getMessage());
        }
    }

    private static void printConfig() {
        logInfo("File Path: " + wifiFilePath);
        logInfo("File Work Path: " + workFilePath);
        logInfo("File Bak Path: " + bakFilePath);
    }

    public static void main(String[] args) {
        // 读取配置
        loadConfig();
        printConfig();

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            logError("Sleep interrupted: " + e.getMessage());
        }

        // 初始化当前日期
        SimpleDateFormat shf = new SimpleDateFormat("HH");

        // 创建工作目录
        try {
            Files.createDirectories(Paths.get(workFilePath));
        } catch (IOException e) {
            logError("Failed to create directory: " + e.getMessage());
        }

        // 初始化H2数据库
        initH2Database();

        // 创建设备表
        createDeviceTable();

        // 创建当天和明天的表
        Calendar tomorrow = Calendar.getInstance();
        tomorrow.add(Calendar.DAY_OF_MONTH, 1);

        // 获取设备字典
        getDeviceDict();

        // 创建线程列表
        List<Thread> threadList = new ArrayList<>();

        // 添加数据文件处理线程
        Thread stThread = new Thread(ObuProcessor::threadST);
        stThread.setDaemon(true);
        stThread.setName("ST_THREAD");
        threadList.add(stThread);

        // 添加解密TEA文件线程
        Thread teaFileDecrypt = new Thread(ObuProcessor::teaFileDecrypt);
        teaFileDecrypt.setDaemon(true);
        teaFileDecrypt.setName("TeaFileDecrypt");
        threadList.add(teaFileDecrypt);

        // 添加解压线程
        Thread unzipThread = new Thread(ObuProcessor::unzipThread);
        unzipThread.setDaemon(true);
        unzipThread.setName("unzip_thread");
        threadList.add(unzipThread);

        Thread fileProcessorThread;

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_RD_1001", ObuProcessor::producerSendJsonRD1001));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_RD_1001");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_RD_1002", ObuProcessor::producerSendJsonRD1002));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_RD_1002");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_RD_1004", ObuProcessor::producerSendJsonRD1004));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_RD_1004");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_RD_1005", ObuProcessor::producerSendJsonRD1005));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_RD_1005");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_WF_1001", null));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_WF_1001");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_WF_1002", null));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_WF_1002");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_WF_1003", null));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_WF_1003");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_BL_1001", null));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_BL_1001");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_BASIC_1001", ObuProcessor::producerSendJsonBasic));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_BASIC_1001");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_STATUS_1001", ObuProcessor::producerSendJsonStatus));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_STATUS_1001");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("ST_SOURCE_DW_1001", null));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("ST_SOURCE_DW_1001");
        threadList.add(fileProcessorThread);

        fileProcessorThread = new Thread(new FileProcessorThread("log", null));
        fileProcessorThread.setDaemon(true);
        fileProcessorThread.setName("log");
        threadList.add(fileProcessorThread);

        // 启动所有线程
        for (Thread t : threadList) {
            t.start();
        }

        // 定时刷新设备缓存
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleAtFixedRate(ObuProcessor::syncDeviceToKafka, 0, 5, TimeUnit.MINUTES);

        String oldHour = shf.format(new Date());

        // 主循环
        while (true) {
            try {
                Thread.sleep(60000); // 60秒

                Date current = new Date();
                String newHourNew = shf.format(current);

                // 检查小时变化
                if (!oldHour.equals(newHourNew)) {
                    getDeviceDict();
                    oldHour = newHourNew;
                }

                // 检查线程状态
                for (Thread t : threadList) {
                    if (t.isAlive()) {
                        continue;
                    }

                    logError(t.getName() + " error");
                    logError("Thread abnormal shutdown, exiting in 10 seconds...");
                    threadExit = -1;
                    Thread.sleep(20000);
                    System.exit(-1);
                }

                // 检查错误数量
                if (errorNo.get() > 30) {
                    threadExit = -1;
                    Thread.sleep(20000);
                    System.exit(-1);
                }
            } catch (InterruptedException e) {
                logError("Main loop interrupted: " + e.getMessage());
                break;
            }
        }

        // 关闭日志
        logger.close();
    }
}
